/*
 * np_socket.c
 *
 *  Created on: Nov 22, 2010
 *      Author: jeff_zheng
 *   Copyright: NetPerform Technology
 */

#include <inttypes.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/select.h>
#include <netinet/in.h>
#include <netinet/ip.h> /* superset of previous */
#include <netdb.h>

#include "np_common.h"
#include "np_socket.h"

struct np_socket_s
{
	np_socket_type			socket_type;
	np_socket_opt	 		opt;

    void*					p_usr_data;
    int         			listen_sock;
    int            			max_fd;
    fd_set         			fdset_backup;
    fd_set         			fdset_sel;
    np_list			    	conn_list;

};


struct np_sock_conn_s
{
    np_list_node		    list_node;
	struct sockaddr_in 		_remote_saddr;
	struct sockaddr_in6 	_remote_saddr6;
	struct sockaddr* 		p_remote_saddr;
	int 					saddr_len;

	np_bool				connected;
	uint32_t                remote_ip;
    uint16_t                remote_port;
    int                     sock;
    np_rng_buf_id           rcv_rng;
    void*                   p_usr_data;
};


np_sock_conn_t* np_sock_conn_create(uint32_t max_buf_size, np_bool create_recv_rng);
int np_sock_conn_close(np_socket* p_tcp, np_sock_conn_t* p_conn);



#define NP_SOCKET_BUF_SIZE_DEFAULT    10240

static int np_sock_conn_data_recv(np_sock_conn_t* p_conn);


#define np_sock_opt_check_msg(condition, fmt, ...) \
	if(condition)	\
	{	\
		NP_ERR(fmt,## __VA_ARGS__); \
		return -1; \
	}

int np_socket_opt_check(np_socket_opt* opts)
{
	switch(opts->socket_type)
	{
	case NP_SOCKET_TYPE_TCP_SERVER:
		np_sock_opt_check_msg(opts->port_num==0, "tcp server: Invalid port number:%u",opts->port_num);
		np_sock_opt_check_msg(opts->o.tcp_svr.single_thread && opts->callback_data_read==NULL,
								"tcp server: single thread model needs a data callback");
		np_sock_opt_check_msg(opts->o.tcp_svr.single_thread && opts->callback_data_read==NULL,
								"tcp server: single thread model needs a data callback");
		np_sock_opt_check_msg((!opts->o.tcp_svr.single_thread) && opts->callback_data_read!=NULL,
								"tcp server: multi thread model doesn't needs a data callback");
		break;
	case NP_SOCKET_TYPE_TCP_CLIENT:
		np_sock_opt_check_msg(opts->port_num==0, "tcp client: Invalid port number:%u",opts->port_num);
		np_sock_opt_check_msg(opts->callback_data_read==NULL, "tcp client: needs a data callback");
		break;
	}

    return 0;
}

int np_socket_init_tcp_client(np_socket* p_socket)
{
    return 0;
}



static int np_socket_bind_tcp_svr(np_socket* p_svr)
{
    struct sockaddr_in local_saddr;
    socklen_t local_saddr_len;
    int ret;

    memset(&local_saddr, 0, sizeof(local_saddr));
    local_saddr.sin_addr.s_addr = INADDR_ANY;
    local_saddr.sin_port = htons(p_svr->opt.port_num);
    local_saddr_len = sizeof(struct sockaddr_in);

    ret = bind(p_svr->listen_sock, (struct sockaddr *)(&local_saddr), local_saddr_len);
    if(ret != 0)
    {
        NP_ERR("bind failed, sock: %d, port_num:%d: %s\n",p_svr->listen_sock, p_svr->opt.port_num, strerror(errno));
        return -1;
    }

    ret = listen(p_svr->listen_sock,1);
    if(ret!=0)
    {
        NP_ERR("listen failed, sock: %d, port_num:%d: %s\n",p_svr->listen_sock, p_svr->opt.port_num, strerror(errno));
        return -1;
    }

    return 0;
}


int np_socket_init_tcp_server(np_socket* p_socket)
{
	int ret, setoptarg;

    p_socket->listen_sock = socket(PF_INET,SOCK_STREAM, 0);
    if(p_socket->listen_sock==-1)
    {
        NP_ERR("create listening socket failed: %s\n",strerror(errno));
        return -1;
    }
    setoptarg=1;
    /*we'll reuse addresses, in case the machine reboot and reconnect quickly*/
    ret = setsockopt(p_socket->listen_sock,SOL_SOCKET,SO_REUSEADDR,&setoptarg,sizeof(setoptarg));
    if(ret!=0)
    {
        NP_LOG("setsockopt return %d %s",ret,strerror(errno));
    }

    if(0!=np_socket_bind_tcp_svr(p_socket))
    {
        NP_ERR("Can not bind to port %u", p_socket->opt.port_num);
        close(p_socket->listen_sock);
        return -1;
    }

    //configure listen sock with select
    FD_SET(p_socket->listen_sock, &(p_socket->fdset_backup));

    p_socket->max_fd = p_socket->listen_sock;
    return 0;
}

const np_socket_opt* np_socket_get_opt(np_socket* p_sock)
{

	return &p_sock->opt;
}


np_socket* np_socket_create(np_socket_opt* opts, void* p_usr_data)
{
    int ret;
    np_socket* p_socket;

    if(np_socket_opt_check(opts)!=0)
    	return NULL;

    p_socket = malloc(sizeof(*p_socket));
	if(p_socket==NULL)
		return NULL;
	memset(p_socket, 0, sizeof(*p_socket));
	p_socket->listen_sock = -1;
	p_socket->max_fd = -1;

	p_socket->p_usr_data = p_usr_data;
	p_socket->opt = *opts;

	/*this buffer size is used by each sock_connection*/
    if(opts->max_buf_size)
    	p_socket->opt.max_buf_size = opts->max_buf_size;
    else
    	p_socket->opt.max_buf_size = NP_SOCKET_BUF_SIZE_DEFAULT;

    if(p_socket->opt.idle_interval.tv_sec==0&&p_socket->opt.idle_interval.tv_usec==0)
    {//default 100ms idel interval
    	p_socket->opt.idle_interval.tv_sec = 0;
    	p_socket->opt.idle_interval.tv_usec = 100000;
    }

	/*initialize main connection list*/
	np_list_init(&(p_socket->conn_list));

	/*initialize select related members*/
	FD_ZERO(&(p_socket->fdset_backup));
	FD_ZERO(&(p_socket->fdset_sel));

	switch(p_socket->opt.socket_type)
	{
	case NP_SOCKET_TYPE_TCP_SERVER:
		ret=np_socket_init_tcp_server(p_socket);
		break;
	case NP_SOCKET_TYPE_TCP_CLIENT:
		ret=np_socket_init_tcp_client(p_socket);
		break;
	default:
		ret = -1;
		break;
	}
	if(ret!=0)
	{
		free(p_socket);
		return NULL;
	}
	return p_socket;
}


int np_socket_check_max_fd(np_socket* p_socket)
{
    int max_fd;
    np_sock_conn_t* p_conn;
    max_fd = p_socket->listen_sock;
    if(p_socket->socket_type==NP_SOCKET_TYPE_TCP_CLIENT|| p_socket->opt.o.tcp_svr.single_thread)
    {  	//only calculate all fds in single thread mode
		np_list_for_each_entry(p_conn, &(p_socket->conn_list), list_node)
		{
			if(max_fd<p_conn->sock)
			{
				max_fd = p_conn->sock;
			}
		}
    }
    p_socket->max_fd = max_fd;
    return 0;
}

int np_socket_tcp_accept_connection(np_socket* p_svr)
{
    int ret;
    struct sockaddr_in peer_saddr;
    socklen_t peer_saddr_len;
    np_sock_conn_t* p_conn;
    int usr_ret;

    np_str_declare(iptmp, 24);

    p_conn = np_sock_conn_create(p_svr->opt.max_buf_size, p_svr->opt.o.tcp_svr.single_thread);
    if(p_conn==NULL)
    {
        return -1;
    }

    peer_saddr_len = sizeof(peer_saddr);
    ret = accept(p_svr->listen_sock, (struct sockaddr *)&peer_saddr, &peer_saddr_len);
    if(ret==-1)
    {
        NP_ERR("accept error: %s",strerror(errno));
        np_sock_conn_close(p_svr, p_conn);
        return -1;
    }

    p_conn->remote_ip = peer_saddr.sin_addr.s_addr;
    p_conn->remote_port = peer_saddr.sin_port;
    p_conn->sock = ret;
    NP_LOG("new tcp connection from %s:%d", np_str_from_ip(&iptmp, 24, p_conn->remote_ip), ntohs(p_conn->remote_port));

    /*get user to know the connection*/
    usr_ret =p_svr->opt.callback_conn_new(p_svr, p_svr->p_usr_data, p_conn);
    if(usr_ret==-1)
	{/*user has either an error, or don't want this connection, close it*/
		np_sock_conn_close(p_svr, p_conn);
		//this is not a lib error, so we set usr_ret to 0 for returning
		usr_ret = 0;
	}

	np_list_add_tail(&(p_conn->list_node),&(p_svr->conn_list));
	if(p_svr->opt.o.tcp_svr.single_thread)
	{
		if(p_svr->max_fd < p_conn->sock)
			p_svr->max_fd = p_conn->sock;
		FD_SET(p_conn->sock, &(p_svr->fdset_backup));
	}
	p_conn->connected = np_true;
    return usr_ret;
}

int np_socket_data_receive(np_socket* p_tcp)
{
    np_sock_conn_t* p_conn, *p_conn_safe;

    NP_INFO("[NP_SOCK]Try receive data");
    int usr_proc_ret;
    np_list_for_each_entry_safe(p_conn, p_conn_safe, &(p_tcp->conn_list), list_node)
    {
        if(FD_ISSET(p_conn->sock, &(p_tcp->fdset_sel)))
        {
        	/*implement fair scheduling, this will cause deadloop
        	 * TODO: Shall put the conn onto a separate list and recat the list to the conn_list.
        	 **/
//        	np_list_del(&p_conn->list_node);
//        	np_list_add_tail(&p_conn->list_node, &(p_tcp->conn_list));

            if(0!=np_sock_conn_data_recv(p_conn))
            {/*peer has closed the connection, tell caller to shutdown the connection*/
            	/*either something has gone wrong, or peer has closed the connection*/
				/*we'll remove this connection from the active connection list*/
				FD_CLR(p_conn->sock, &(p_tcp->fdset_backup));
				np_list_del(&(p_conn->list_node));
				np_sock_conn_close(p_tcp, p_conn);
				np_socket_check_max_fd(p_tcp);
				return -1;
            }
            usr_proc_ret = p_tcp->opt.callback_data_read(p_tcp, p_tcp->p_usr_data, p_conn, p_conn->p_usr_data, p_conn->rcv_rng);
            if(usr_proc_ret<0)
            {/*error in user data processing, tell caller to close connection*/
            	NP_ERR("[SOCK][%"PRIx64"] Error when processing received data from connection ", (uintptr_t)p_conn);
				FD_CLR(p_conn->sock, &(p_tcp->fdset_backup));
				np_list_del(&(p_conn->list_node));
				np_sock_conn_close(p_tcp, p_conn);
				np_socket_check_max_fd(p_tcp);
				sleep(1);
                return -1;
            }
            if(usr_proc_ret>0)
            {//user want to interrupt receiving
            	return 1;
            }
        }
    }
    return 0;
}

int np_socket_idle_process(np_socket* p_tcp)
{
    np_sock_conn_t* p_conn, *p_conn_safe;
    if(p_tcp->opt.callback_conn_idle==NULL)
    	return 0;
    np_bool breakloop = np_false;
    np_list_for_each_entry_safe(p_conn, p_conn_safe, &(p_tcp->conn_list), list_node)
	{
    	int ret = p_tcp->opt.callback_conn_idle(p_tcp, p_tcp->p_usr_data, p_conn, p_conn->p_usr_data);
    	if(ret==1)
    	{
    		breakloop = np_true;
    	}
    	else if(ret==-1)
    	{//user want to shutdown the connection
			np_list_del(&(p_conn->list_node));
			np_sock_conn_close(p_tcp, p_conn);
			np_socket_check_max_fd(p_tcp);
    	}
	}
    if(breakloop)
    {
    	return 1;
    }
    return 0;
}

int np_socket_data_recv_loop(np_socket* p_tcp)
{
    struct timeval sel_tmo;

    int ret;
    int usr_ret;

    while(p_tcp->max_fd>0)
    {
    	//we need to restore the whole fdset everytime, from a backup copy
        memcpy(&p_tcp->fdset_sel,&p_tcp->fdset_backup,sizeof(p_tcp->fdset_backup));
        sel_tmo.tv_sec = p_tcp->opt.idle_interval.tv_sec;
        sel_tmo.tv_usec = p_tcp->opt.idle_interval.tv_usec;
        ret = select(p_tcp->max_fd + 1, &(p_tcp->fdset_sel), NULL, NULL, &sel_tmo);
        if(ret==0)
        {/*timeout*/
        	if(np_socket_idle_process(p_tcp)==1)
        		break;
            continue;
        }
        if(ret<0)
        {/*error*/
            NP_ERR("select error: %s\n",strerror(errno));
            //do a sleep here to prevent the thread to use 100% CPU
            np_time_sleep(0, 1000000);
            continue;
        }
        if(p_tcp->listen_sock>0)
        {//only valid for a server
			if(FD_ISSET(p_tcp->listen_sock,  &(p_tcp->fdset_sel)))
			{/*new connection*/
				FD_CLR(p_tcp->listen_sock,  &(p_tcp->fdset_sel));
				usr_ret = np_socket_tcp_accept_connection(p_tcp);
	        	if(usr_ret>0)
	        	{//user want to interrupt receiving
	        		return 1;
	        	}
			}
        }
        if(p_tcp->socket_type==NP_SOCKET_TYPE_TCP_CLIENT||p_tcp->opt.o.tcp_svr.single_thread)
        {
        	usr_ret=np_socket_data_receive(p_tcp);
        	if(usr_ret>0)
        	{//user want to interrupt receiving
        		return 1;
        	}
        }

    }
    return 0;

}

np_sock_conn_t* np_sock_conn_create(uint32_t max_buf_size, np_bool create_recv_rng)
{
    np_sock_conn_t* p_conn;
    p_conn = malloc(sizeof(*p_conn));
    if(p_conn==NULL)
        return NULL;
    memset(p_conn, 0, sizeof(*p_conn));
//    NP_TRACE("[SOCK][%"PRIx64"] Creating socket connection ", (uintptr_t)p_conn);

    p_conn->sock = -1;
    p_conn->connected = np_false;

    if(create_recv_rng)
    {
		p_conn->rcv_rng= np_rng_buf_create(1, max_buf_size, NULL);
		if(p_conn->rcv_rng==NULL)
		{
			free(p_conn);
			return NULL;
		}
    }
    return p_conn;
}

int np_sock_conn_close(np_socket* p_tcp, np_sock_conn_t* p_conn)
{
	int usr_ret;
//    np_str_declare(iptmp, 24);
    np_socket_check_max_fd(p_tcp);

//    NP_TRACE("[SOCK][%"PRIx64"] Closing socket connection ", (uintptr_t)p_conn);
	usr_ret = p_tcp->opt.callback_conn_close(p_tcp, p_tcp->p_usr_data, p_conn, p_conn->p_usr_data);

    if(p_conn->sock>=0)
    {
        shutdown(p_conn->sock, SHUT_RDWR);
        close(p_conn->sock);
//        NP_TRACE("[SOCK][%"PRIx64"] Close TCP connection from %s:%d", (uintptr_t)p_conn,
//        				np_str_from_ip(&iptmp, 24, p_conn->remote_ip), ntohs(p_conn->remote_port));
    }
    if(p_tcp->socket_type==NP_SOCKET_TYPE_TCP_SERVER||p_tcp->opt.o.tcp_svr.single_thread)
    {
    	np_rng_buf_destroy(p_conn->rcv_rng);
    	p_conn->rcv_rng = 0;
    }

    free(p_conn);
    return usr_ret;
}

void np_sock_conn_set_arg(np_sock_conn_t* p_conn, void* p_usr_arg)
{
	p_conn->p_usr_data = p_usr_arg;
	return;
}

int np_sock_conn_get_sock(np_sock_conn_t* p_conn)
{
	return p_conn->sock;
}

int np_sock_conn_send(np_sock_conn_t* p_conn, void* buffer, uint32_t size)
{
    int ret;
    ret = send(p_conn->sock, buffer, size, MSG_NOSIGNAL);
    if(ret==(int)size)
        return 0;
    if(ret<0)
    {
    	NP_ERR("[SOCK][%"PRIx64"] vd_sock_write err: %s", (uintptr_t)p_conn, strerror(errno));
        return -1;
    }
    return 0;
}

/*return -1, if there is error and socket need to be closed,
  return 0 otherwise*/
int np_sock_conn_data_recv(np_sock_conn_t* p_conn)
{
    int ret;
    uint32_t size;
    uint8_t buf[2048];

    do
    {
        size = np_rng_buf_get_free_count(p_conn->rcv_rng);
        if(size==0)
        {/*not enough buffer space for holding received data*/
        	return 0;
        }
        if(size > sizeof(buf))
        	size = sizeof(buf);
		ret = recv(p_conn->sock, buf, size, MSG_DONTWAIT);
		if(ret<0)
		{
			int err;
			err = errno;
			if(errno==ENOTCONN || errno == ENOTSOCK || errno == EBADF)
			{
	        	NP_ERR("[SOCK][%"PRIx64"] Error on receive %s", (uintptr_t)p_conn, strerror(errno));
				return -1;
			}
			if(err==EAGAIN)
			{
				NP_INFO("[SOCK][%"PRIx64"] EAGAIN on receive", (uintptr_t)p_conn);
				return 0;
			}
		}
		else if(ret==0)
		{/*peer has shutdown*/
			NP_INFO("[SOCK][%"PRIx64"] Peer has closed connection", (uintptr_t)p_conn);
			return -1;
		}
		else
		{
//        	NP_LOG("[SOCK][%"PRIx64"] Received %10u bytes", (uintptr_t)p_conn, ret);
			np_rng_buf_push(p_conn->rcv_rng, buf, ret);
		}
    }while(ret>0);

    return 0;
}



int np_tcp_client_conn_setup(np_socket* p_client, np_sock_conn_t* p_conn)
{
	struct hostent host, *p_host = NULL;
	int h_err;
	char tmpbuf[1024];

	gethostbyname_r(p_client->opt.o.tcp_client.remote_host, &host, tmpbuf, sizeof(tmpbuf), &p_host, &h_err);
	if(p_host==NULL)
	{
    	NP_LOG("[SOCK][%"PRIx64"] Failed to resolve %s", (uintptr_t)p_conn, p_client->opt.o.tcp_client.remote_host);
		return -1;
	}

	if(p_host->h_addrtype==AF_INET)
	{
		p_conn->_remote_saddr.sin_family = AF_INET;
		p_conn->_remote_saddr.sin_addr.s_addr = *(in_addr_t*)(p_host->h_addr_list[0]);
		p_conn->_remote_saddr.sin_port = htons(p_client->opt.port_num);
		p_conn->p_remote_saddr = (struct sockaddr*)&p_conn->_remote_saddr;
		p_conn->saddr_len = sizeof(p_conn->_remote_saddr);
	}
	else if(p_host->h_addrtype==AF_INET6)
	{
		p_conn->_remote_saddr6.sin6_family = AF_INET6;
		memcpy(p_conn->_remote_saddr6.sin6_addr.s6_addr, p_host->h_addr_list[0], p_host->h_length);
		p_conn->_remote_saddr6.sin6_port = htons(p_client->opt.port_num);
		p_conn->p_remote_saddr = (struct sockaddr*)&p_conn->_remote_saddr6;
		p_conn->saddr_len = sizeof(p_conn->_remote_saddr);
	}

	return 0;
}

int np_tcp_client_connect(np_socket* p_client, np_sock_conn_t* p_conn)
{
	int ret = 0;
    p_conn->sock = socket(PF_INET,SOCK_STREAM, 0);
    if(p_conn->sock==-1)
    {
    	NP_ERR("[SOCK][%"PRIx64"] Failed to create socket: %s", (uintptr_t)p_conn, strerror(errno));
        return -1;
    }
	do
	{
		ret=connect(p_conn->sock, p_conn->p_remote_saddr, p_conn->saddr_len);
		if(ret==0)
		{
			int usr_ret = p_client->opt.callback_conn_new(p_client, p_client->p_usr_data, p_conn);
			if(usr_ret==1)
				return 1;
			if(usr_ret==-1)
			{
				NP_LOG("[SOCK][%"PRIx64"] Caller rejected this connection ", (uintptr_t)p_conn);
				close(p_conn->sock);
				p_conn->sock=-1;
				return -1;
			}
			np_list_add_tail(&(p_conn->list_node), &(p_client->conn_list));
			FD_SET(p_conn->sock, &(p_client->fdset_backup));
			np_socket_check_max_fd(p_client);
			p_conn->connected = np_true;
			return 0;
		}
		else{
	    	NP_LOG("[SOCK][%"PRIx64"] Failed to connect to %s(%u) %s", (uintptr_t)p_conn,
	    				p_client->opt.o.tcp_client.remote_host, p_client->opt.port_num, strerror(errno));
	    	sleep(1);//sleep for 1 second before making another connection attempt
		}
	}
	while(p_client->opt.o.tcp_client.reconnect);

	return ret;
}

int np_socket_tcp_client_loop(np_socket* p_client)
{
	int recv_ret;
    do
    {
        np_sock_conn_t* p_conn;
        if(np_list_empty(&(p_client->conn_list)))
        {//we don't have a np_sock_conn_t structure yet
        	p_conn = np_sock_conn_create(p_client->opt.max_buf_size, np_true);
        }
        else
        {
        	p_conn = np_list_entry(p_client->conn_list.next, np_sock_conn_t, list_node);
        }
        if(p_conn==NULL)
        {
        	return -1;
        }
        if(!p_conn->connected)
        {
        	int usr_ret;
        	np_tcp_client_conn_setup(p_client, p_conn);

			usr_ret=np_tcp_client_connect(p_client, p_conn);
			if(usr_ret == 1)
				return 1;
			else if(usr_ret==-1)
			{
				//failed to connect, close the current connection
				np_sock_conn_close(p_client, p_conn);
				if(p_client->opt.o.tcp_client.quit_on_disconnect)
					break;
				sleep(1);//sleep for 1 second before making another connection attempt
				continue;
			}
        }
		//do a data loop, to do all the communication
		recv_ret=np_socket_data_recv_loop(p_client);
		if(recv_ret<0)
		{
			np_sock_conn_close(p_client, p_conn);
		}
		if(recv_ret==1)
		{//user want to interrupt receive, give control back to the caller
			return 1;
		}
    }while(!p_client->opt.o.tcp_client.quit_on_disconnect);

    return 0;
}

int np_socket_client_close_curr_conn(np_socket* p_client)
{
    np_sock_conn_t* p_conn;
    int usr_ret = 0;
    if(!np_list_empty(&(p_client->conn_list)))
    {
    	p_conn = np_list_entry(p_client->conn_list.next, np_sock_conn_t, list_node);
    	np_list_del(&p_conn->list_node);
    	if(p_client->opt.callback_conn_close)
    	{
    		p_client->opt.callback_conn_close(p_client, p_client->p_usr_data, p_conn, p_conn->p_usr_data);
    	}
    	usr_ret = np_sock_conn_close(p_client, p_conn);
    }
    return 0;
}
